package com.alex.interceptors;

import java.util.Map;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

/**
 * @ClassName UserDefineProducerInterceptor
 * @Description TODO
 * @Author zjg
 * @Date 2021/12/710:13
 * @Version 2.0
 */

public class UserDefineProducerInterceptor implements ProducerInterceptor {

  private volatile  long sendSucess = 0;

  private volatile  long sendFail =0;

  @Override
  public ProducerRecord onSend(ProducerRecord record) {
    return new ProducerRecord(record.topic(),record.key(),record.value()+"---alex.test");
  }

  @Override
  public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
    if(e==null){
      sendSucess++;
    }else{
      sendFail++;
    }
    System.out.println("recordMetadata:"+recordMetadata+",exception:"+e);
  }

  @Override
  public void close() {

    System.out.println("--发送成功数：--"+sendSucess+",    发送失败数：--"+sendFail+"--");
    double successRation =(double)sendSucess/(sendSucess+sendFail);
    System.out.println("--[INFO] 发送成功率："+String.format("%f",successRation*100)+"%");
  }

  @Override
  public void configure(Map<String, ?> map) {
    System.out.println("--configure--");
  }
}
